//
// Created by Eric Zhu on 2021/1/31.
//
#include <iostream>
#include <chrono>
#include "LogSender.hpp"
#include "../Common/SocketHelper.hpp"

namespace cybertron {
	LogSender::LogSender(const std::string& server, uint16_t port, SafeQueue<NetPacket> *send_queue)
		: _server(server)
		, _port(port)
		, _send_queue(send_queue) {

	}

	void LogSender::run_once() {
		if (!SocketHelper::is_valid_address(_server.c_str())) {
			//std::cout << _server << " is not a valid address\n";
			std::this_thread::sleep_for(std::chrono::milliseconds(100));
			return;
		}

		union_endpoint local_ep;
		SOCKET s = SocketHelper::socket(local_ep);

		if (s == -1) {
			std::cout << "socket error\n";
			std::this_thread::sleep_for(std::chrono::milliseconds(100));
			return;
		}

		SocketHelper::set_blocking(s, false);

		union_endpoint remote_ep(_server.c_str(), _port);
		auto start = ::time(NULL);
		int err = SocketHelper::connect(s, remote_ep);

		if (err != 0 && !SocketHelper::again()) {
			//std::cout << SocketHelper::last_error() << "\n";
			//std::cout << "connect error" << "\n";
			std::this_thread::sleep_for(std::chrono::milliseconds(100));
			return;
		}

		LogPacketSender *io = new LogPacketSender(s, remote_ep, _send_queue);
		_io_manager.add_io(io);

		while (!_quit_flag) {
			if (!io->connected()) {
				auto end = ::time(NULL);
				if (end - start > 10) {
					std::cout << "connection timed out" << "\n";
					io->on_close();
					break;
				}
			}

			std::unordered_set<SocketIO *> error_set;
			_io_manager.perform_io(NULL, &error_set, 10);

			if (!error_set.empty()) {
				break;
			}
		}
		_io_manager.del_io(io);
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}

	int LogPacketSender::on_read()
	{
		char buf[128];
		int ret = 0;

		do {
			ret = SocketHelper::recv(_socket, buf, 128);
		} while (ret > 0);

		if (SocketHelper::again()) {
			return 0;
		}
		else {
			return -1;
		}
	}

	int LogPacketSender::on_write()
	{
		if (!_connected) {
			//std::cout << "connected" << "\n";
			_connected = true;
		}

		_ongoing_packet = _ongoing_packet?_ongoing_packet:_send_queue->dequeue();
		while (_ongoing_packet) {
			size_t to_send = _ongoing_packet->send_buf_len();
			int ret = SocketHelper::send(_socket, _ongoing_packet->send_buf(), to_send);

			if (ret > 0) {
				_ongoing_packet->commit_send(ret);
				if (_ongoing_packet->all_sent()) {
					delete _ongoing_packet;
					_ongoing_packet = _send_queue->dequeue();
				}
				if (to_send == ret) {
					continue;
				}
				else {
					return 0;
				}
			}
			else if (SocketHelper::again()) {
				return 0;
			}
			else {
				return -1;
			}
		}

		return 0;
	}
}